package org.example.hadoop3;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class MapJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
    private Map<String, String> goodsMap = new HashMap<>();
    @Override
    protected void setup(Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        context.getCounter("Map Join", "setup").increment(1);
        URI[] cacheFiles = context.getCacheFiles();
        if (cacheFiles.length >= 1) {
            URI file = cacheFiles[0];
            FileSystem fs = FileSystem.get(context.getConfiguration());
            FSDataInputStream inputStream = fs.open(new Path(file));
            BufferedReader br = new BufferedReader(new InputStreamReader(inputStream, "utf-8"));
            String line;
            while ((line = br.readLine()) != null) {
                String[] strs = line.split("\t");
                goodsMap.put(strs[0], strs[1]);
            }
            IOUtils.closeStream(br);
        }
    }

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, NullWritable>.Context context) throws IOException, InterruptedException {
        context.getCounter("Map Join", "map").increment(1);
        String line = value.toString();
        String[] strs = line.split("\t");
        strs[1] = goodsMap.get(strs[1]);
        //context.write(StringUtils.join(Arrays.asList(strs)), NullWritable.get());
    }
}
